package validator

import (
	"context"
	"github.com/ecodeclub/ekit/slice"
	"github.com/ecodeclub/ekit/syncx/atomicx"
	"go.uber.org/zap"
	"golang.org/x/sync/errgroup"
	"gorm.io/gorm"
	"jk-time/webook/pkg/migrator"
	"jk-time/webook/pkg/migrator/events"
	"time"
)

type Validator[T migrator.Entity] struct {
	baseValidator
	batchSize int
	producer  events.Producer
	highLoad  *atomicx.Value[bool]
	utime     int64
	// <= 0 就认为中断
	// > 0 就认为睡眠
	sleepTime time.Duration
}

func NewValidator[T migrator.Entity](bv baseValidator) *Validator[T] {
	return &Validator[T]{baseValidator: bv, batchSize: 10, highLoad: atomicx.NewValueOf(false), sleepTime: time.Second}
}
func (v *Validator[T]) SleepInterval(i time.Duration) *Validator[T] {
	v.sleepTime = i
	return v
}
func (v *Validator[T]) Utime(utime int64) *Validator[T] {
	v.utime = utime
	return v
}
func (v *Validator[T]) Validate(ctx context.Context) error {
	var eg errgroup.Group
	eg.Go(func() error {
		return v.ValidateBaseToTarget(ctx)
	})
	eg.Go(func() error {
		return v.ValidateTargetToBase(ctx)
	})
	return eg.Wait()
}
func (v *Validator[T]) ValidateBaseToTarget(ctx context.Context) error {
	offset := 1
	for {
		var src T
		dbCtx, cancel := context.WithTimeout(ctx, time.Second)
		err := v.base.WithContext(dbCtx).Where("utime>=?", v.utime).Order("id").
			Offset(offset).First(&src).Error
		cancel()
		switch err {
		case context.Canceled, context.DeadlineExceeded:
			return nil
		case nil:
			//本表存在
			var dst T
			dbCtx, cancel = context.WithTimeout(ctx, time.Second)
			err1 := v.target.WithContext(dbCtx).Where("id=?", src.ID()).First(&dst).Error
			cancel()
			switch err1 {
			case nil:
				//查到了数据
				ok := dst.CompareTo(dst)
				if !ok {
					v.notify(ctx, src.ID(), events.InconsistentEventTypeNotEqual)
				}
			case gorm.ErrRecordNotFound:
				// 目标少了数据
				v.notify(ctx, src.ID(), events.InconsistentEventTypeTargetMissing)
			default:
				zap.L().Error("src => dst 查询目标表失败:", zap.Error(err),
					zap.Int("offset", offset))
			}
		case gorm.ErrRecordNotFound:
			//没数据了
			//注意 Where("utime>?", v.utime).
			//utime=0且sleep<=0 全量校验退出
			//utime=0且sleep>0 全量校验,增量校验
			//utime>0且sleep>0 校验近期数据,增量校验
			if v.sleepTime <= 0 {
				return nil
			}
			time.Sleep(v.sleepTime)
			continue
		default:
			zap.L().Error("src => dst 查询源表失败:", zap.Error(err),
				zap.Int("offset", offset))
		}
		offset++
		continue
	}
}
func (v *Validator[T]) ValidateBaseToTargetV1(ctx context.Context) error {
	offset := 0
	for {
		var srcTs []T
		dbCtx, cancel := context.WithTimeout(ctx, time.Second)
		err := v.base.WithContext(dbCtx).Where("utime>=?", v.utime).Order("id").
			Offset(offset).Limit(v.batchSize).First(&srcTs).Error
		cancel()
		switch err {
		case context.DeadlineExceeded, context.Canceled:
			return nil
		case nil:
			ids := slice.Map(srcTs, func(idx int, src T) int64 {
				return src.ID()
			})
			var dstTs []T
			dbCtx, cancel = context.WithTimeout(ctx, time.Second)
			err1 := v.target.WithContext(dbCtx).Where("id in ?", ids).Find(&dstTs).Error
			cancel()
			if err1 != nil {
				zap.L().Error("src => dst 查询目标表失败:", zap.Error(err),
					zap.Int("offset", offset))
				return err1
			}
			dstMap := v.toMap(dstTs)
			for _, src := range srcTs {
				dst, ok := dstMap[src.ID()]
				if !ok {
					v.notify(ctx, dst.ID(), events.InconsistentEventTypeTargetMissing)
					continue
				}
				if !src.CompareTo(dst) {
					v.notify(ctx, src.ID(), events.InconsistentEventTypeNotEqual)
				}
			}
		case gorm.ErrRecordNotFound:
			return nil
		default:
			zap.L().Error("src => dst 查询源表失败:", zap.Error(err),
				zap.Int("offset", offset))
			return err
		}
		offset += v.batchSize
		if len(srcTs) < v.batchSize {
			return nil
		}
		continue
	}
}

// A utime = 昨天
// A base 今天被删了，target 数据还在，utime还是昨天，那今天恢复的就不行了
func (v *Validator[T]) ValidateTargetToBase(ctx context.Context) error {
	// 先找target,再找base,找到base中已经被删除
	offset := 0
	for {
		var dstTs []T
		dbCtx, cancel := context.WithTimeout(ctx, time.Second)
		err := v.target.WithContext(dbCtx).Where("utime>=?", v.utime).Order("id").
			Offset(offset).Limit(v.batchSize).Find(&dstTs).Error
		cancel()
		switch err {
		case context.DeadlineExceeded, context.Canceled:
			return err
		case gorm.ErrRecordNotFound:
			//没数据了
			//utime=0且sleep<=0 全量校验退出
			//utime=0且sleep>0 全量校验,增量校验
			//utime>0且sleep>0 校验近期数据,增量校验
			if v.sleepTime <= 0 {
				return nil
			}
			time.Sleep(v.sleepTime)
			continue
		case nil:
			//目标表存在
			ids := slice.Map(dstTs, func(idx int, src T) int64 {
				return src.ID()
			})
			var srcTs []T
			dbCtx, cancel = context.WithTimeout(ctx, time.Second)
			err := v.base.WithContext(dbCtx).Where("id in ?", ids).Find(&srcTs).Error
			cancel()
			switch err {
			case gorm.ErrRecordNotFound:
				v.notifySrcMissing(ctx, dstTs)
			case nil:
				// 计算差集
				missing := slice.DiffSetFunc(dstTs, srcTs, func(src, dst T) bool {
					return src.ID() == dst.ID()
				})
				v.notifySrcMissing(ctx, missing)
			default:
				zap.L().Error("dst => src 查询源表失败:", zap.Error(err),
					zap.Int("offset", offset))
			}
		default:
			zap.L().Error("dst => src 查询目标表失败:", zap.Error(err),
				zap.Int("offset", offset))
		}
		offset += v.batchSize
		if len(dstTs) < v.batchSize {
			//没数据了
			//同时支持全量校验和增量校验，就不能退出
			if v.sleepTime <= 0 {
				return nil
			}
			time.Sleep(v.sleepTime)
		}
	}

}
func (v *Validator[T]) notifySrcMissing(ctx context.Context, ts []T) {
	for _, t := range ts {
		v.notify(ctx, t.ID(), events.InconsistentEventTypeBaseMissing)
	}
}

func (v *Validator[T]) toMap(ts []T) map[int64]T {
	mp := make(map[int64]T, len(ts))
	for _, val := range ts {
		mp[val.ID()] = val
	}
	return mp
}
